package com.fanshuai.spring.boot.config;

import com.fanshuai.spring.boot.component.GenericKafkaListener;
import com.fanshuai.spring.boot.component.KafkaConsumerEngine;
import com.fanshuai.spring.boot.component.handler.MessageHandler;
import com.fanshuai.spring.boot.component.listener.ConsumerListener;
import com.fanshuai.spring.boot.interceptor.MyConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Configuration
public class KafkaConsumerConfig {
    @Value("${spring.kafka.consumer.bootstrap-servers}")
    private String servers;

    @Value("${spring.kafka.consumer.group-id}")
    private String groupId;

    @Autowired(required = false)
    private List<MessageHandler> handlerList;
    @Autowired(required = false)
    private List<ConsumerListener> listeners;

    @Bean
    @Primary
    public ConsumerFactory consumerFactory() {
        return new DefaultKafkaConsumerFactory<String, String>(props());
    }

    private Map<String, Object> props() {
        Map<String, Object> map = new HashMap<String, Object>();
        map.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        map.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        map.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        map.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        map.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        map.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, MyConsumerInterceptor.class.getName());

        return map;
    }

    @ConditionalOnBean(MessageHandler.class)
    @Bean
    public GenericKafkaListener kafkaListener(ConsumerFactory<String, String> consumerFactory) {
        return new GenericKafkaListener(handlerList, listeners, consumerFactory);
    }

    @Bean(destroyMethod = "close")
    public KafkaConsumerEngine kafkaConsumerEngine(GenericKafkaListener kafkaListener, ConsumerFactory<String, String> consumerFactory) {
        return new KafkaConsumerEngine(consumerFactory, handlerList, kafkaListener);
    }
}
